package com.bigdata;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;

/**
 * @className: LogETLInterceptor
 * @description: TODO 通过flume拦截器实现数据清洗过滤

 *
 *flume拦截器一：ETL
 *flume采集到的数据，首先会经过该拦截器进行过滤，对每一条数据进行数据校验，将不合法的数据过滤掉；
 *
 *实现flume自定义拦截器步骤：
 * 1）实现拦截器接口：Interceptor
 * 2）实现方法：
 *     （1）初始化方法：initialize() , 只执行一次，资源的申请
 *     （2）系统自调用方法：intercept(List<Event> events) ，flume会把接收到的一个批次的数据封装到一个集合中，
 *                     然后调用该方法进行业务处理；
 *     （3）业务逻辑方法：intercept(Event event) ， 具体来执行拦截器的业务逻辑；
 *     （4）释放资源方法：close() 只执行一次
 * 3）生成拦截器对象，从而实现拦截器的业务功能：
 *    通过拦截器接口的子接口 Interceptor.Builder ，来实现自定义拦截器对象的实例化（系统调用，如果自己构建一个对象，系统
 *    无法获取；）
 *    （1）实例化方法：build() ， 构建拦截器对象
 *    （2）拦截器配置方法：configure(Context context) ： 可以添加一些参数
 */
public class LogETLInterceptor implements Interceptor {

    //初始化方法：只执行一次
    @Override
    public void initialize() { }

    /**
     * @Param events:
     * @return java.util.List<org.apache.flume.Event>
     * @description TODO 系统调用方法：flume将采集到的数据封装到一个批次中，然后调用拦截器，校验完返回一个批次：
     */
    @Override
    public List<Event> intercept(List<Event> events) {
        ArrayList<Event> interceptors = new ArrayList<>();
        // 在遍历的过程中，对source发送来的每一条数据进行业务逻辑处理：
        for (Event event : events) {
            //数据合法校验：
            Event resultEvent = this.intercept(event);
            if (resultEvent != null){
                interceptors.add(resultEvent);
            }
        }
        return interceptors;
    }

    /**
     * @Param event:
     * @return org.apache.flume.Event
     * @description TODO 业务逻辑方法：对于flume来说，是基于事件机制的，会把每一条数据当做一个event事件来处理：一行文本内容被反序列化成一个event。event的最大定义为2048字节，超过会被切割放到下一个event中。
     */
    @Override
    public Event intercept(Event event) {
        //1、获取一条数据：
        byte[] body = event.getBody();
        String log = new String(body, Charset.forName("UTF-8"));
        //2、判断一条的数据类型:
        if (log.contains("start")) {
            //启动日志数据合法校验：
            if (LogUtils.validateStart(log)){
                return event;
            }
        }else {
            //事件日志数据合法校验：
            if (LogUtils.validateEvent(log)){
                return event;
            }
        }
        //3、如果都不满足，返回null:
        return null;
    }

    //关闭资源：只执行一次
    @Override
    public void close() { }

    /**
     * 功能：构建外部类对象
     */
    public static class Builder implements Interceptor.Builder{
        // 拦截器对象实例化：
        @Override
        public Interceptor build() {
            return new LogETLInterceptor();
        }
        // 拦截器参数配置：
        @Override
        public void configure(Context context) { }
    }
}
